package com.sugon.ohdfs.integration.flink.job.stream.source;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import com.sugon.ohdfs.integration.flink.format.parquet.AvroParquetReaders;
import com.sugon.ohdfs.integration.flink.job.stream.sink.BulkParquetStreamSinkJob;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.impl.StreamFormatAdapter;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.core.fs.Path;
import org.springframework.beans.factory.annotation.Value;

public class ParquetBulkStreamSourceJob extends AbstractStreamSourceJob{

    @Value("${hdfs.base-path:hdfs://10.11.8.29:9001/flink/sinkTest/stream/sink/bulk/parquet}")
    String basePath;

    @Override
    protected FileSource buildFileSource() {
        AvroParquetReaders.forReflectRecord(TestItem.class);
        BulkFormat bulkFormat = new StreamFormatAdapter(AvroParquetReaders.forReflectRecord(TestItem.class));
        return FileSource
                .forBulkFileFormat(bulkFormat, new Path(basePath))
                .build();
    }

    @Override
    protected TestItem transform(Object item) {
        BulkParquetStreamSinkJob.TestItemParquet testItemParquet = (BulkParquetStreamSinkJob.TestItemParquet) item;
        return testItemParquet.toTestItem();
    }
}
